package cn.boat.router.utils;

import lombok.Data;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

//import org.apache.commons.logging.Log;
//import org.apache.commons.logging.LogFactory;

/**
 * Zookeeper工具类（Curator）
 */
public class ZkCuratorUtils {
    //private static Log logger = LogFactory.getLog(ZkClientUtils.class);

    private static final String DEFAULT_CONNECT_IP = "127.0.0.1:2181";
    private static final int DEFAULT_SESSION_TIME_OUT = 5000;
    private static final int DEFAULT_CONNECTION_TIME_OUT = 3000;
    private static final RetryPolicy DEFAULT_RETRY_POLICY = new ExponentialBackoffRetry(1000, 3);
    private static Map<String, NodeCache> pathListeners = new ConcurrentHashMap<>();
    private static Map<String, PathChildrenCache> pathChiidrenListeners = new ConcurrentHashMap<>();

    private static Map<String, TreeCache> nodeListeners = new ConcurrentHashMap<String, TreeCache>();


    public static CuratorFramework getCuratorFramework(){
        return getCuratorFramework(DEFAULT_CONNECT_IP);
    }

    public static CuratorFramework getCuratorFramework(String IP){
        return getCuratorFramework(IP, DEFAULT_SESSION_TIME_OUT, DEFAULT_CONNECTION_TIME_OUT, DEFAULT_RETRY_POLICY);
    }

    public static CuratorFramework getCuratorFramework(String IP, Integer sessionTimeOut, Integer connectionTimeOut, RetryPolicy retryPolicy){
        CuratorFramework curator = CuratorFrameworkFactory.builder()
                .connectString(IP)
                .sessionTimeoutMs(sessionTimeOut)
                .connectionTimeoutMs(connectionTimeOut)
                .retryPolicy(retryPolicy)
                .build();
        return curator;
    }

    /**
     * 创建持久无序节点
     * @param client Curator
     * @param path 路径
     */
    public static void createPersistentNode(CuratorFramework client, String path) throws Exception {
        createPersistentNode(client, path, null);
    }

    /**
     * 创建持久无序节点
     * @param client Curator
     * @param path 路径
     * @param data 数据
     */
    public static void createPersistentNode(CuratorFramework client, String path, String data) throws Exception {
        createNode(client, path, data, true, CreateMode.PERSISTENT, ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 创建持久有序节点
     * @param client
     * @param path
     */
    public static void createPersistentSequentialNode(CuratorFramework client, String path) throws Exception {
        createPersistentSequentialNode(client, path, null);
    }
    /**
     * 创建持久有序节点
     * @param client
     * @param path
     * @param data
     */
    public static void createPersistentSequentialNode(CuratorFramework client, String path, String data) throws Exception {
        createNode(client, path, data, true, CreateMode.PERSISTENT_SEQUENTIAL,ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 创建父路径
     * @param client Curator
     * @param path 当前路径
     */
    public static void createParentPath(CuratorFramework client, String path) throws Exception {
        createParentPath(client, path, null);
    }

    /**
     * 创建父路径
     * @param client Curator
     * @param path 当前路径
     * @param data 数据
     */
    public static void createParentPath(CuratorFramework client, String path, String data) throws Exception {
        String parentPath = path.substring(0, path.lastIndexOf('/'));
        if (parentPath.length() != 0) {
            createNode(client, parentPath, data, true, CreateMode.PERSISTENT, ZooDefs.Ids.OPEN_ACL_UNSAFE);
        }
    }

    /**
     * 创建临时无序节点
     * @param client Curator
     * @param path 路径
     */
    public static void createEphemeralNode(CuratorFramework client, String path) throws Exception {
        createEphemeralNode(client, path, null);
    }

    /**
     * 创建持久无序节点
     * @param client Curator
     * @param path 路径
     * @param data 数据
     */
    public static void createEphemeralNode(CuratorFramework client, String path, String data) throws Exception {
        createNode(client, path, data, true, CreateMode.EPHEMERAL, ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 创建临时有序节点
     * @param client
     * @param path
     */
    public static void createEphemeralSequentialNode(CuratorFramework client, String path) throws Exception {
        createEphemeralSequentialNode(client, path, null);
    }
    /**
     * 创建临时有序节点
     * @param client
     * @param path
     * @param data
     */
    public static void createEphemeralSequentialNode(CuratorFramework client, String path, String data) throws Exception {
        createNode(client, path, data, true, CreateMode.EPHEMERAL_SEQUENTIAL, ZooDefs.Ids.OPEN_ACL_UNSAFE);
    }

    /**
     * 创建节点
     * @param client Curator
     * @param path 路径
     * @param data 数据
     * @param createParent 是否创建父路径
     * @param mode 持久模型
     * @param acl 执行权限
     */
    public static void createNode(CuratorFramework client, String path, String data, boolean createParent,
                                  CreateMode mode, ArrayList<ACL> acl) throws Exception {
        if (createParent) {

            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(mode)
                    .withACL(acl)
                    .forPath(path, data==null?null:data.getBytes());

        }else{
            client.create()
                    .withMode(mode)
                    .withACL(acl)
                    .forPath(path, data==null?null:data.getBytes());
        }
    }

    /**
     * 删除节点
     * @param client
     * @param path
     */
    public static void deleteNode(CuratorFramework client, String path) throws Exception {
        deleteNode(client, path, false, -1);
    }

    /**
     * 删除节点及其子节点
     * @param client
     * @param path
     */
    public static void deleteNodeAndChildren(CuratorFramework client, String path) throws Exception {
        deleteNode(client, path, true, -1);
    }

    /**
     * 删除节点
     * @param client Curator
     * @param path 路径
     * @param deleteChildren 是否一并删除子节点
     * @param version 版本号
     */
    public static void deleteNode(CuratorFramework client, String path, boolean deleteChildren, int version) throws Exception {
        if (deleteChildren) {
            client.delete()
                    .deletingChildrenIfNeeded()
                    .withVersion(version)
                    .forPath(path);
        }else{
            client.delete()
                    .withVersion(version)
                    .forPath(path);
        }
    }

    /**
     * 修改数据
     * @param client Curator
     * @param path 路径
     * @param data 是否一并删除子节点
     */
    public static void updateNode(CuratorFramework client, String path, String data){
        updateNode(client, path, data, -1);
    }

    /**
     * 修改数据
     * @param client Curator
     * @param path 路径
     * @param data 是否一并删除子节点
     * @param version 版本
     */
    public static void updateNode(CuratorFramework client, String path, String data, int version){
        try {
            client.setData().withVersion(version).forPath(path, data==null?null:data.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取路径对应值
     * @param path 路径
     * @return 值
     * @throws Exception
     */
    public static String getNodeData(CuratorFramework client, String path) throws Exception {
        byte[] bytes = client.getData().forPath(path);
        return bytes==null?null:new String(bytes);
    }


    /**
     * 获取路径stat信息
     * @param client Curator
     * @param path 路径
     * @return Stat
     * @throws Exception
     */
    public static Stat getNodeStat(CuratorFramework client, String path) throws Exception {
        Stat stat = new Stat();
        client.getData().storingStatIn(stat).forPath(path);
        return stat;
    }

    /**
     * 获取所有子节点
     * @param client Curator
     * @param path 路径
     * @return
     * @throws Exception
     */
    public static List<String> getChildren(CuratorFramework client, String path) throws Exception {
        List<String> list = client.getChildren().forPath("/get");
        return list;
    }

    /**
     * 检查是否存在
     * @param client Curator
     * @param path 路径
     * @return Stat（存在） / Null（不存在时）
     * @throws Exception
     */
    public static Stat checkIfExists(CuratorFramework client, String path) throws Exception {
        Stat stat= client.checkExists().forPath(path);
        return stat;
    }

    /**
     * // 监视某个节点的数据变化
     * @param client Curator
     * @param path 路径
     * @param listener 监听器
     */
    public static void addPathListener(CuratorFramework client, String path, NodeCacheListener listener) throws Exception {
        final NodeCache nodeCache=new NodeCache(client, path);
        nodeCache.start();
        pathListeners.put(path, nodeCache);
        nodeCache.getListenable().addListener(listener);
    }

    /**
     * 增加某个节点的监听
     * @param path
     * @param callback
     * @throws Exception
     */
    public static void addListener(CuratorFramework client, String path , Consumer<TreeCacheEvent> callback) throws Exception {
        TreeCache treeCache = new TreeCache(client, path);
        nodeListeners.put(path, treeCache);
        treeCache.start();
        treeCache.getListenable().addListener((client1, event) -> {
            try {
                callback.accept(event);
            } finally {
                if(event.getType() == TreeCacheEvent.Type.NODE_REMOVED) {
                    String path1 = event.getData().getPath();
                    if(path1 !=null) {
                        // 移除监听器
                        removeListener(path1);
                    }
                }
            }
        });
    }
    /**
     * 删除此节点上的监听
     * @param path
     */
    private static void removeListener(String path) {
        TreeCache treeCache = nodeListeners.remove(path);
        if(treeCache!=null) {
            treeCache.close();
        }
    }

    /**
     * 移除某个节点的监听器
     * @param path 路径
     * @throws IOException
     */
    public static void removePathListener(String path) throws IOException {
        if (pathListeners.containsKey(path)) {
            pathListeners.get(path).close();
            pathListeners.remove(path);
        }
    }

    /**
     * 监视某个节点子节点的变化
     * @param client Curator
     * @param path 路径
     * @param cacheData 事件中是否可以获取节点的数据
     * @param listener 监听器
     * @throws Exception
     */
    public static void addPathChildrenListener(CuratorFramework client, String path, boolean cacheData, PathChildrenCacheListener listener) throws Exception {
        PathChildrenCache pathChildrenCache=new PathChildrenCache(client, path, cacheData);
        // 启动监听
        pathChildrenCache.start();
        pathChildrenCache.getListenable().addListener(listener);
        pathChiidrenListeners.put(path, pathChildrenCache);
    }

    /**
     * 移除某个节点子节点的监听器
     * @param path 路径
     * @throws IOException
     */
    public static void removePathChildrenListener(String path) throws IOException {
        if (pathChiidrenListeners.containsKey(path)) {
            pathChiidrenListeners.get(path).close();
            pathChiidrenListeners.remove(path);
        }
    }

    @Data
    public static class ZKConfig extends Config implements Serializable {
        static final long serialVersionUID = -1L;

        /**
         * 根
         */
        @Key(name = "zk.zkRoot")
        public String zkRoot = "/boatMQ";

        /**
         * zookeeper主机
         */
        @Key(name = "zk.zkConnect")
        public String zkConnect;

        /**
         * zookeeper会话超时时间（默认30秒）
         */
        @Key(name = "zk.zkSessionTimeoutMs")
        public int zkSessionTimeoutMs = 30000;

        /**
         * 连接zookeeper最大容忍时间（默认30秒）
         */
        @Key(name = "zk.zkConnectionTimeoutMs")
        public int zkConnectionTimeoutMs = 30000;

        /**
         * zookeeper主从之间传输间隔时间
         */
        @Key(name = "zk.zkSyncTimeMs")
        public int zkSyncTimeMs = 5000;

        public ZKConfig() {
            setZkConnect("127.0.0.1:2181");
        }

        public ZKConfig(final String zkConnect, final int zkSessionTimeoutMs, final int zkConnectionTimeoutMs,
                        final int zkSyncTimeMs) {
            super();
            this.zkConnect = zkConnect;
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
            this.zkSyncTimeMs = zkSyncTimeMs;
        }

        public ZKConfig(final String zkRoot, final String zkConnect, final int zkSessionTimeoutMs,
                        final int zkConnectionTimeoutMs, final int zkSyncTimeMs) {
            super();
            this.zkRoot = zkRoot;
            this.zkConnect = zkConnect;
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
            this.zkSyncTimeMs = zkSyncTimeMs;
        }

        public static long getSerialVersionUID() {
            return serialVersionUID;
        }

        public String getZkRoot() {
            return zkRoot;
        }

        public void setZkRoot(String zkRoot) {
            this.zkRoot = zkRoot;
        }

        public String getZkConnect() {
            return zkConnect;
        }

        public void setZkConnect(String zkConnect) {
            this.zkConnect = zkConnect;
        }

        public int getZkSessionTimeoutMs() {
            return zkSessionTimeoutMs;
        }

        public void setZkSessionTimeoutMs(int zkSessionTimeoutMs) {
            this.zkSessionTimeoutMs = zkSessionTimeoutMs;
        }

        public int getZkConnectionTimeoutMs() {
            return zkConnectionTimeoutMs;
        }

        public void setZkConnectionTimeoutMs(int zkConnectionTimeoutMs) {
            this.zkConnectionTimeoutMs = zkConnectionTimeoutMs;
        }

        public int getZkSyncTimeMs() {
            return zkSyncTimeMs;
        }

        public void setZkSyncTimeMs(int zkSyncTimeMs) {
            this.zkSyncTimeMs = zkSyncTimeMs;
        }
    }

    /**
     * 从zookeeper.properties加载
     * @return ZKConfig
     */
    public ZKConfig loadZkConfigFromLocalFile() {
        try {
            Properties properties = PropUtils.getResourceAsProperties("zookeeper.properties", "GBK");
            ZKConfig zkConfig = new ZKConfig();
            if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnect"))) {
                zkConfig.zkConnect = properties.getProperty("zk.zkConnect");
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkSessionTimeoutMs"))) {
                zkConfig.zkSessionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkSessionTimeoutMs"));
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkConnectionTimeoutMs"))) {
                zkConfig.zkConnectionTimeoutMs = Integer.parseInt(properties.getProperty("zk.zkConnectionTimeoutMs"));
            }

            if (StringUtils.isNotBlank(properties.getProperty("zk.zkSyncTimeMs"))) {
                zkConfig.zkSyncTimeMs = Integer.parseInt(properties.getProperty("zk.zkSyncTimeMs"));
            }

            return zkConfig;
        } catch (final IOException e) {
            //log.error("zk配置失败", e);
            System.out.println("zookeeper配置失败！");
            return null;
        }
    }
}
